package com.atguigu.flink.window;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.stream.Stream;
import java.util.stream.StreamSupport;

/**
 * Created by Smexy on 2023/2/27
 *
 *  1.13 中如果使用的是基于个数的滑动窗口，
 *      此时 滚动聚合算子将失效，并且是累积聚合！
 *
 *      有这种窗口下的聚合需求，就使用process,自己编写聚合逻辑！
 */
public class Demo9_Bug
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //求每种传感器的水位和
        env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .keyBy(WaterSensor::getId)
            .countWindow(5,3)
            //等窗口关闭后，才会把数据发往下游的print
            .reduce(new ReduceFunction<WaterSensor>()
            {
                @Override
                public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {

                    System.out.println("Demo6_Reduce.reduce");
                    //正常
                    //value1.setVc(value1.getVc() + value2.getVc());
                    //累积
                    value2.setVc(value1.getVc() + value2.getVc());
                    return value2;
                }
            })
            /*.process(new ProcessWindowFunction<WaterSensor, String, String, GlobalWindow>()
            {
                @Override
                public void process(String key, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {

                    //集合都可以使用StreamAPI操作
                    Stream<WaterSensor> stream = StreamSupport.stream(elements.spliterator(), true);

                    int sumVC = stream.mapToInt(w -> w.getVc()).sum();

                    out.collect(key + ":" + sumVC);

                }
            })*/
            .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
